package com.briup.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * mr程序的默认组件  以及默认的输出内容
 */

public class DefaultMR extends Configured  implements Tool {


    @Override
    public int run(String[] args) throws Exception {
//          获取配置对象
        Configuration conf = getConf();
        String in = conf.get("in");
        String out = conf.get("out");
//        创建 Job对象
        Job job = Job.getInstance(conf);
//        给job起名字 等基础配置
        job.setJobName("defaultMR");
        job.setJarByClass(this.getClass());
//获取默认的输入处理器  InputFormat  TextInputFormat
        Class<? extends InputFormat<?, ?>> inputFormatClass = job.getInputFormatClass();
//        获取默认的Mapper类    Mapper
        Class<? extends Mapper<?, ?, ?, ?>> mapperClass = job.getMapperClass();
//        获取默认的Partitioner类  HashPartitioner
        Class<? extends Partitioner<?, ?>> partitionerClass = job.getPartitionerClass();
//        获取默认的Reducer   Reducer
        Class<? extends Reducer<?, ?, ?, ?>> reducerClass = job.getReducerClass();
//        获取默认的输出处理器 TextOutputFormat
        Class<? extends OutputFormat<?, ?>> outputFormatClass = job.getOutputFormatClass();
//        获取默认的reduceTask个数  1
        int numReduceTasks = job.getNumReduceTasks();
        System.out.println("inputFormatClass = " + inputFormatClass);
        System.out.println("mapperClass = " + mapperClass);
        System.out.println("partitionerClass = " + partitionerClass);
        System.out.println("reducerClass = " + reducerClass);
        System.out.println("outputFormatClass = " + outputFormatClass);
        System.out.println("numReduceTasks = " + numReduceTasks);

//        不自定义Mapper和Reducer  进行数据输入和输出
        Path inPath = new Path(in);
        Path outPath = new Path(out);
//        告诉输入处理器  给哪一个输入作业设置输入路径 以及要输入的文件路径
        TextInputFormat.addInputPath(job,inPath);
//      告诉job k2,v2的类型
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setNumReduceTasks(2);
//       告诉job k3,v3的类型
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);
//      告诉输出处理器  给哪一个作业设置输出路径 以及要输出的文件路径
        TextOutputFormat.setOutputPath(job,outPath);
//        提交job
        return job.waitForCompletion(true)?0:-1;
    }


    //    程序执行的主入口
    public static void main(String[] args) throws Exception {
//        执行MapReduce程序
        ToolRunner.run(new DefaultMR(),args);

    }

}


